package com.ziroom.crm.mq.core.processor.retry;

import com.ziroom.crm.mq.core.config.MqConfig;
import com.ziroom.crm.mq.core.context.InvokeContext;
import com.ziroom.crm.mq.core.processor.ConsumeProcessor;
import com.ziroom.crm.mq.core.processor.QueueWrapper;
import com.ziroom.crm.mq.core.retry.*;
import com.ziroom.crm.mq.core.spi.SpiLoader;
import com.ziroom.crm.mq.core.spi.SpiOrder;
import com.ziroom.crm.mq.core.util.AssertUtil;
import lombok.extern.slf4j.Slf4j;

import java.util.Date;
import java.util.Objects;
import java.util.function.Consumer;

/**
 * <p>重试机制</p>
 * 重试的不是具体实现，而是reConsume,reSend 等
 * <p>
 * <PRE>
 * <BR>    修改记录
 * <BR>-----------------------------------------------
 * <BR>    修改日期         修改人          修改内容
 * </PRE>
 *
 * @author WeiShaoying
 * @version 1.0
 * @date Created in 2020年09月08日 10:06
 * @since 1.0
 */
@SpiOrder(SpiOrder.LOWEST_PRECEDENCE - 1)
@Slf4j
public class DefaultConsumeRetryProcessor implements ConsumeProcessor {

    public static final String GROUP_CONSUME = "consume";

    private final RetryHandler retryHandler;


    public DefaultConsumeRetryProcessor() {
        // spi 方式加载指定配置
        String retryPolicyName = MqConfig.getConfig(MqConfig.RETRY_CONSUME_POLICY);
        String delayPolicyName = MqConfig.getConfig(MqConfig.RETRY_CONSUME_DELAY_POLICY);
        //log.debug("retryPolicyName = " + retryPolicyName + ",delayPolicyName = " + delayPolicyName);

        RetryPolicy retryPolicy = SpiLoader.loadPrototypeInstanceByAlias(RetryPolicy.class, retryPolicyName, GROUP_CONSUME);
        AssertUtil.notNull(retryPolicy, "RetryPolicy with alias[" + retryPolicyName + "] and group [" + GROUP_CONSUME + "] must has one implement!");
        DelayPolicy delayPolicy = SpiLoader.loadPrototypeInstanceByAlias(DelayPolicy.class, delayPolicyName, GROUP_CONSUME);
        AssertUtil.notNull(retryPolicy, "DelayPolicy with alias[" + delayPolicyName + "] and group [" + GROUP_CONSUME + "] must has one implement!");

        retryHandler = new DefaultRetryHandler(retryPolicy, delayPolicy);
    }


    @Override
    public void postConsume(InvokeContext invokeContext, QueueWrapper queueWrapper) {
        Consumer<RetryContext> callback = queueWrapper.getRetryCallback();
        if (Objects.isNull(callback)) {
            return;
        }
        if (Objects.isNull(invokeContext.getMessage().getException())) {
            return;
        }
        // 发生异常，重试
        Integer initRetryCount = invokeContext.getMessage().getRetryCount();
        // 默认内存重试
        RetryContext retryContext = RetryContext.valueOf(invokeContext, RetryType.IMMEDIATELY, initRetryCount);
        RetryFunction retryFunction = callback::accept;
        retryHandler.retry(retryFunction, retryContext);
        // 设置持久化后的重试时间
        if (Objects.nonNull(retryContext.getMessage().getException())) {
            setNextTime(retryContext);
        }
    }

    protected void setNextTime(RetryContext retryContext) {
        // 10分钟后
        Integer span = MqConfig.getConfigInt(MqConfig.RETRY_CONSUME_SECOND_BEFORE_RECONSUME, 0);
        retryContext.getMessage().setNextRetryDate(new Date(System.currentTimeMillis() + span * 1000));
    }

}
